package com.ndp.fb.mq.service.schedulerservice;

import com.alibaba.dubbo.config.annotation.Service;
import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.mq.api.service.schedulerservice.FetchRetentionFilesScheduleService;
import com.ndp.fb.mq.api.service.schedulerservice.retention.RetentionRecordService;
import com.ndp.fb.util.DateFormatUtil;
import com.ndp.fb.util.DateUtil;
import com.ndp.fb.util.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.*;
import java.util.*;

/**
 * Created by elvisxiong on 10/9/15.
 */
@Service
public class FetchRetentionFilesScheduleServiceImpl implements FetchRetentionFilesScheduleService {

    private static Logger logger = LoggerFactory.getLogger(FetchRetentionFilesScheduleServiceImpl.class);
    @Autowired
    private RetentionRecordService retentionRecordService;
    private int downloadOffsetDay = -1;

    /**
     * 获取s3客户端实例
     *
     * @return
     */
    private AmazonS3 getS3Instance() {

        String accessKeyId = ConfigCentre.getString(ConfigConst.S3_RETENTION_RATE_ACCESS_KEY_ID);
        String secretAccessKey = ConfigCentre.getString(ConfigConst.S3_RETENTION_RATE_SECRET_ACCESS_KEY);
        AmazonS3 s3 = new AmazonS3Client(new BasicAWSCredentials(accessKeyId, secretAccessKey));
        com.amazonaws.regions.Region usWest2 = com.amazonaws.regions.Region.getRegion(Regions.US_EAST_1);
        s3.setRegion(usWest2);
        return s3;
    }

    /**
     * 删除临时文件
     *
     * @param fileName 文件名
     */
    private void deleteTmpFile(String fileName) {

        String tmpPath = ConfigCentre.getString(ConfigConst.S3_RETENTION_RATE_TEMP_PATH);
        File file = new File(tmpPath + "/" + fileName);
        if (file.exists()) {
            file.delete();
        }
    }

    @Override
    public void fetchProcessFiles() {

        logger.info("进入留存文件下载服务.");
        //获取今天之前的一天日期
        downloadOffsetDay = ConfigCentre.getInteger(ConfigConst.S3_RETENTION_DOWNLOAD_OFFSET_DAY)*-1; //配置中心配正整数
        String suffix = DateFormatUtil.formatShort8(DateUtil.addDateByTypeAndOffset(new Date(),Calendar.DATE, downloadOffsetDay));

        //设置4个文件
        Map hashMap = new HashMap<>();
        hashMap.put("fmp_" + suffix + ".txt", 0);
        hashMap.put("facebook_" + suffix + ".txt", 1);
        hashMap.put("organic_" + suffix + ".txt", 2);
        hashMap.put("ad_" + suffix + ".txt", 3);

        //遍历4个文件，启动4个线程进行处理
        Iterator iter = hashMap.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry entry = (Map.Entry) iter.next();
            final String key = (String) entry.getKey();
            final int val = (int) entry.getValue();
            Runnable threadRuable = new Runnable() {
                @Override
                public void run() {
                    logger.info("s3:key:" + key, "val:" + val);
                    analysisFileToDb(key, val);
                    deleteTmpFile(key);
                }
            };
            ThreadPool.execute(threadRuable);
        }
    }

    /**
     * 分析文件并存入数据库，下载文件失败，休息3秒钟后，重试5次如果仍然失败放弃
     *
     * @param fileName 文件名
     * @param fileType 0 应用留存数据，1精准到ad的留存数据
     */
    private void analysisFileToDb(String fileName, int fileType) {
        logger.info("开始获取fileName:" + fileName);
        boolean flag = true;
        int i = 0;
        File tmpFmpFile = null;
        while (flag && i < 5) {
            try {
                logger.info(fileName + "第" + (i + 1) + "次下载");
                tmpFmpFile = awsS3FetchFile(fileName);
                if (tmpFmpFile != null) {
                    logger.info("己下载文件" + fileName + "到临时目录。");
                } else {
                    logger.error(fileName + "不存在。");
                }
                flag = false;
            } catch (Exception ex) {
                logger.error("aws s3获取文件失败,", ex);
            }
            try {
                Thread.sleep(3000l);
            } catch (InterruptedException e) {
                logger.error("线程休眠异常," + e);
            }
            i++;
        }
        try {
            if (tmpFmpFile != null) {
                if (fileType == 0) {
                    logger.info("开始保存应用统计信息至mongdb,文件名:" + tmpFmpFile.getName());
                    retentionRecordService.saveApp(tmpFmpFile,"fmp");
                } else if (fileType == 1) {
                    logger.info("开始保存应用统计信息至mongdb,文件名:" + tmpFmpFile.getName());
                    retentionRecordService.saveApp(tmpFmpFile, "facebook");
                }else if (fileType == 2) {
                    logger.info("开始保存应用统计信息至mongdb,文件名:" + tmpFmpFile.getName());
                    retentionRecordService.saveApp(tmpFmpFile, "organic");
                }else if (fileType == 3) {
                    logger.info("开始保存ad统计信息至mongdb,文件名:" + tmpFmpFile.getName());
                    retentionRecordService.saveAds(tmpFmpFile);
                }
            }
        } catch (FileNotFoundException e) {
            logger.error("文件:" + fileName + "没有找到异常,", e);
        }
    }

    /**
     * 通过aws-client获取存储到aws s3上的文件
     *
     * @param fileName 文件名
     * @return 下载到临时目录的文件
     */
    private File awsS3FetchFile(String fileName) throws Exception {
        File file = null;
        AmazonS3 s3 = null;
        String bucketName = ConfigCentre.getString(ConfigConst.S3_RETENTION_RATE_BUCKET_NAME);
        String tmpPath = ConfigCentre.getString(ConfigConst.S3_RETENTION_RATE_TEMP_PATH);
        try {
            s3 = getS3Instance();
            logger.info(fileName + "开始调用s3客户端。");
            GetObjectRequest getObjectRequest = new GetObjectRequest(bucketName, "bigdata_stat/"+fileName);
            S3Object s3Object = s3.getObject(bucketName, "bigdata_stat/"+fileName);
            file = new File(tmpPath + "/" + fileName);
            s3.getObject(getObjectRequest, file);
            if (!file.isDirectory() && file.length() != s3Object.getObjectMetadata().getContentLength()) {
                throw new Exception();
            }
        } catch (AmazonServiceException ase) {
            logger.error("获取aws服务异常," + ase);
        } catch (AmazonClientException ace) {
            logger.error("获取aws客户端异常," + ace);
        } catch (Exception e) {
            logger.error("其它异常错误，" + e);
        } finally {
            s3 = null;
        }
        return file;
    }
}
